package com.zyx.flinkdemo.stream.watermark;

import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;

import java.time.Duration;

import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;

/**
 * @author Yaxi.Zhang
 * @since 2021/5/27 17:30
 * desc:
 */
public class MyWatermarkGenerator<T> implements WatermarkGenerator<T> {
    private long maxTimestamp;
    private final long outOfOrdernessMillis;
    /**
     * 是否有必要对周期性触发Watermark的次数进行增加
     */
    private boolean isEmit;
    /**
     * 没有数据进入时周期性Watermark的触发次数
     */
    private int monotonousTimes;

    public MyWatermarkGenerator(Duration maxOutOfOrderness) {
        checkNotNull(maxOutOfOrderness, "maxOutOfOrderness");
        checkArgument(!maxOutOfOrderness.isNegative(), "maxOutOfOrderness cannot be negative");
        this.outOfOrdernessMillis = maxOutOfOrderness.toMillis();
        this.maxTimestamp = Long.MIN_VALUE + outOfOrdernessMillis + 1;
        this.isEmit = false;
    }

    @Override
    public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
        monotonousTimes = 0;
        isEmit = true;
        maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
    }

    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        // 没有数据进入时
        if (isEmit) {
            monotonousTimes++;
        }
        if (monotonousTimes > 25) {
            isEmit = false;
            output.emitWatermark(new Watermark(maxTimestamp - outOfOrdernessMillis + 5000 - 1));
        } else {
            output.emitWatermark(new Watermark(maxTimestamp - outOfOrdernessMillis - 1));
        }
    }
}